package com.daqian.chance.elasticsearch.web.modules.elasticsearch.controller;

import com.alibaba.fastjson.JSONObject;
import com.daqian.chance.common.base.BaseController;
import com.daqian.chance.core.rest.RestResult;
import com.daqian.chance.elastic.search.service.impl.ElasticSearchService;
import com.daqian.chance.elasticsearch.web.modules.elasticsearch.po.es.GoodsDoc;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.bulk.BackoffPolicy;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;

import java.math.BigDecimal;
import java.util.function.BiConsumer;


@Slf4j
@RestController
@RequestMapping("/test/elasticSearch")
public class ElasticSearchController extends BaseController {

    @Autowired
    private ElasticSearchService elasticSearchService;

    private final static String TEST_INDEX_NAME = "goods_info";

    /**
     * 插入/修改文档
     */
    @PostMapping("index/document")
    public RestResult createDocument() {
        GoodsDoc goodsDoc = new GoodsDoc();
        goodsDoc.set_id("1");
        goodsDoc.setId(1L);
        goodsDoc.setName("name test");
        goodsDoc.setPrice(new BigDecimal("33.13"));
        goodsDoc.setTag("tag test");
        goodsDoc.setSaleCount(12);
        elasticSearchService.insertOrUpdateDoc(TEST_INDEX_NAME, goodsDoc);
        return RestResult.ok();
    }

    /**
     * 查询文档
     */
    @GetMapping("index/document/{docId}")
    public RestResult searchDocument(@PathVariable("docId") Integer docId) {
        // TODO: 2019/12/17
        return RestResult.ok();
    }


    /**
     * 删除文档
     */
    @DeleteMapping("index/document/{docId}")
    public RestResult deleteDocument(@PathVariable("docId") Integer docId) {
        // TODO: 2019/12/17
        return RestResult.ok();
    }


    /**
     * 分页查询 用户-信息
     * @author daqian
     */
    @PostMapping("update/{indexName}/{len}")
    public RestResult page(@PathVariable("indexName") String indexName, @PathVariable("len") int len, @RequestBody String dataStr) {
//        baseElasticSearchService.insertOrUpdateOne(indexName, data);
        System.out.println("xxxx len是" + len);
        for (int i = 0; i < len; i++) {
            BulkProcessor add = bulkProcessor.add(new IndexRequest(indexName).source(JSONObject.parseObject(dataStr).getInnerMap()));
            System.out.println(add);
        }
        System.out.println("完毕！！！！");
        return RestResult.ok();
    }

    @Autowired
    private RestHighLevelClient restHighLevelClient;

    private BulkProcessor.Listener bulkProcessorListener = new BulkProcessor.Listener() {
        @Override
        public void beforeBulk(long executionId, BulkRequest request) {
            int numberOfActions = request.numberOfActions();
            log.debug("Executing bulk [{}] with {} requests", executionId, numberOfActions);
        }

        @Override
        public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
            if (response.hasFailures()) {
                log.warn("Bulk [{}] executed with failures", executionId);
            } else {
                log.debug("Bulk [{}] completed in {} milliseconds", executionId, response.getTook().getMillis());
            }
        }

        @Override
        public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
            log.error("Failed to execute bulk", failure);
        }
    };

    private BiConsumer<BulkRequest, ActionListener<BulkResponse>> bulkConsumer =
            (request, bulkListener) ->
                    restHighLevelClient.bulkAsync(request, RequestOptions.DEFAULT, bulkListener);

    private BulkProcessor bulkProcessor =
            BulkProcessor.builder(bulkConsumer, bulkProcessorListener)
                    //根据当前添加的操作数设置何时刷新批量请求（默认为1000，使用-1禁用）
//            .setBulkActions(500) // todo:
                    .setBulkActions(3)
                    //设置何时根据当前添加的操作大小刷新批量请求（默认为5Mb，使用-1禁用它）
                    .setBulkSize(new ByteSizeValue(1L, ByteSizeUnit.MB))
                    //设置允许执行的并发请求数（默认为1，使用0只允许执行单个请求）
                    .setConcurrentRequests(0)
                    //设置时间间隔：如果超过，则设置刷新间隔刷新任何BulkRequest挂起（默认为未设置）
                    .setFlushInterval(TimeValue.timeValueSeconds(10L))
                    //设置指数退避算法策略，如果一个服务提供者的服务在某一时间发生了异常、超时或是网络抖动,则最多重试3次。
                    .setBackoffPolicy(BackoffPolicy
                            .constantBackoff(TimeValue.timeValueSeconds(1L), 3)).build();

}
